/**
 * Copyright (c) 2011-2013, kidzhou 周磊 (zhouleib1412@gmail.com)
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.jfinal.ext.plugin.jms;

import com.jfinal.ext.plugin.jms.hanlder.DefaultMessageReceive;
import com.jfinal.ext.plugin.jms.hanlder.IMessageHandler;
import com.jfinal.log.Log;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;


public class TopicListener implements MessageListener {

    protected final Log logger = Log.getLog(getClass());

    protected String serverUrl;
    protected String username;
    protected String password;
    protected String topicName;

    private Connection connection;
    private Session session;
    private Destination destination;
    private MessageConsumer consumer;
    private IMessageHandler messageHandler;


    public TopicListener(String serverUrl, String username, String password, String topicName,
            IMessageHandler messageHandler) {
        this.serverUrl = serverUrl;
        this.username = username;
        this.password = password;
        this.topicName = topicName;
        this.messageHandler = messageHandler;
        initListener();
    }


    private void initListener() {
        try {
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(serverUrl);
            connection = connectionFactory.createConnection(username, password);
//            connection.setClientID("Topic-consumer");
            connection.start();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //create topic
            //5.11使用destination = new ActiveMQTopic(topicName);
            destination = session.createTopic(topicName);
            consumer = session.createConsumer(destination);
//            consumer = session.createDurableSubscriber(session.createTopic(topicName),topicName);
            consumer.setMessageListener(this);
        }
        catch (JMSException e) {
            logger.error("init error", e);
        }
    }


    @Override public void onMessage(Message message) {
        if (messageHandler != null) {
            try {
                logger.info("call onMessage:" + ((ObjectMessage) message).getObject().toString());
            }
            catch (JMSException e) {
                e.printStackTrace();
            }
            try {
                messageHandler.handleMessage(message);
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
        else {
            logger.error("MessageHandler is null!please set a messageHandler");
            new DefaultMessageReceive().handleMessage(message);
        }
    }


    public void closeConnection() {
        if (consumer != null) {
            try {
                consumer.close();
            }
            catch (JMSException e) {
                logger.error(e.getMessage(), e);

            }
        }
        if (session != null) {
            try {
                session.close();
            }
            catch (JMSException e) {
                logger.error(e.getMessage(), e);

            }
        }
        if (connection != null) {
            try {
                connection.close();
            }
            catch (JMSException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
}
